package com.example.mq.mqclient;

import com.example.mq.common.*;
import org.apache.ibatis.reflection.ExceptionUtil;
import org.springframework.http.converter.feed.RssChannelHttpMessageConverter;

import javax.xml.crypto.dsig.CanonicalizationMethod;
import java.awt.image.RescaleOp;
import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Connection {

    private Socket socket = null;

    // 管理多个 channel
    private ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>();

    private InputStream inputStream;

    private OutputStream outputStream;

    private DataInputStream dataInputStream;

    private DataOutputStream dataOutputStream;

    private ExecutorService callbackPool = null;

    public Connection(String host, int port) throws IOException {
        socket = new Socket(host, port);
        inputStream = socket.getInputStream();
        outputStream = socket.getOutputStream();
        dataInputStream = new DataInputStream(inputStream);
        dataOutputStream = new DataOutputStream(outputStream);

        callbackPool = Executors.newFixedThreadPool(4);

        // 创建一个扫描线程，由这个线程不停的从 socket 中读取响应
        // 把这个响应数据在交给对应的 channel 来处理
        Thread t = new Thread(()->{
            try{
                while (!socket.isClosed()){
                    Response response = readResponse();
                    dispatchResponse(response);
                }
            }catch (SocketException e){
                System.out.println("[Connection] 连接正常断开！");
            }catch(IOException | ClassNotFoundException | MqException e){
                System.out.println("[Connection] 连接异常断开！");
                e.printStackTrace();
            }
        });
        t.start();
    }

    // 关闭 Connection 释放上述资源
    public void close() throws IOException {
        try{
            callbackPool.shutdownNow();
            channelMap.clear();
            inputStream.close();
            outputStream.close();
            socket.close();
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    // 使用这个方法用来分别处理，当前的请求是一个针对控制请求的响应还是服务器推送的消息
    private void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
        if(response.getType() == 0xc){
            // 服务器推送来的消息数据
            SubScribeReturns subScribeReturns = (SubScribeReturns) BinaryTool.fromBytes(response.getPlayload());
            // 根据 channelId 找到对应的channel 对象
            Channel channel = channelMap.get(subScribeReturns.getChannelId());
            if(null == channel){
                throw new MqException("[Connection] 该消息对应的 channel 在客户端不存在！ channelId = " + channel.getChannelId());
            }
            // 执行该 channel 对象内部的消息回调
            callbackPool.submit(()->{
               try{
                   channel.getConsumer().handleDelivery(subScribeReturns.getConsumerTag(), subScribeReturns.getBasicProperties(),
                           subScribeReturns.getBody());
               }catch (MqException | IOException e){
                   e.printStackTrace();
               }
            });
        }else{
            // 当前响应是针对刚才的控制请求的响应
            BasicReturns basicReturns = (BasicReturns) BinaryTool.fromBytes(response.getPlayload());
            // 把这个结果放到对应的 channel  的 hash 表中
            Channel channel = channelMap.get(basicReturns.getChannelId());
            if(null == channel){
                throw new MqException("[Connection] 该消息对应的 channel 在客户端中不存在！ channelId = " + channel.getChannelId());
            }
            channel.putReturns(basicReturns);
        }
    }

    // 发送请求
    public void writeRequest(Request request) throws IOException {
        dataOutputStream.writeInt(request.getType());
        dataOutputStream.writeInt(request.getLength());
        dataOutputStream.write(request.getPayload());
        dataOutputStream.flush();
        System.out.println("[Connection] 发送请求！ type = " + request.getType() + ", length = " + request.getLength());
    }

    // 读取响应
    public Response readResponse() throws IOException {
        Response response = new Response();
        response.setType(dataInputStream.readInt());
        response.setLength(dataInputStream.readInt());
        byte[] payload = new byte[response.getLength()];
        int n = dataInputStream.read(payload);
        if(n != response.getLength()){
            throw new IOException("读取的响应数据不完整！");
        }
        response.setPlayload(payload);
        System.out.println("[Connection] 收到响应！ type = " + response.getType() + ", length = " + response.getLength());
        return response;
    }

    // 在 Connection 中创建一个 channel1
    public Channel createChannel() throws IOException {
        String channelId = "C-" + UUID.randomUUID().toString();
        Channel channel = new Channel(channelId, this);
        // 把 channel 对象放到 Connection 管理的 channel 的哈希表中
        channelMap.put(channelId, channel);
        // 将创建 channel 的消息告诉服务器
        boolean ok = channel.createChannel();
        if(!ok){
            // 创建失败，将加入哈希表中的 channel 删除
            channelMap.remove(channelId);
            return null;
        }
        return channel;
    }
}
